package com.sjc.lesson02.api.streaming.transformation;

import com.sjc.lesson02.api.streaming.source.function.MyNoParalleSource;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 数据源：1 2 3 4 5.....源源不断过来
 * 通过map打印一下接受到数据
 * 通过filter过滤一下数据，我们只需要偶数
 */
public class MapDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Long> numberStream = env.addSource(new MyNoParalleSource()).setParallelism(1);

        SingleOutputStreamOperator<Long> dataStream =
                numberStream.map(new MapFunction<Long, Long>() {
                    @Override
                    public Long map(Long value) throws Exception {
                        System.out.println("接收到了数据：" + value);
                        return value;
                    }
                });

        SingleOutputStreamOperator<Long> filterDataStream =
                dataStream.filter(new FilterFunction<Long>() {
                    @Override
                    public boolean filter(Long number) throws Exception {
                        return number % 2 == 0;
                    }
                });

        filterDataStream.print().setParallelism(1);
        env.execute("MapDemo");
    }
}
